package com.xiaoyilin.fast;

import com.fasterxml.jackson.databind.deser.std.StringDeserializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * Kafka中的消费是基于拉模式
 * 对于 poll（）方法而言，如果某些分区中没有可供消费的消息，那么此分区对应的消息拉取的 结果就为空 ；
 * 如果订阅的所有分区中都没有可供消费的消息，那么 poll（）方法返回为空的消息集
 * <p>
 * ConsumerRebalanceListener再均衡监昕器
 * public void subscribe(Collection<String> topics ,ConsumerRebalanceListener listener)
 * <p>
 * public void subscribe(Collection<String> topics)
 * public void subscribe (Pattern pattern , ConsumerRebalanceListener listener) public void subscribe (Pattern pattern)
 * <p>
 * 正则订阅
 * consumer.subscribe (Pattern.compile (”topic.*”))
 * <p>
 * 消费者不仅可以通过 KafkaConsumer.subscribe（）方法订阅主题，
 * 还可直接订阅某些主题的特定分区 ，在KafkaConsumer 中还提供了 assign（）方法来实现这些功能
 * <p>
 * org.apache.kafka.common.TopicPartition
 * public void assign(Collection<TopicPartition> partitions)
 * <p>
 * KafkaConsumer 的 partitionsFor（）方法可以用来查询指定主题的元数据信息，
 * Partitionlnfo类即为主题的分区元数据信息  replicas 代表分区的 AR 集合， inSyncReplicas 代表分区 的 ISR 集合， off lineReplicas 代表分区的 OSR 集合 。
 * <p>
 * KafkaConsumer 中的 unsubscribe（）方法采取消 主题的订阅。
 * 这个方法既可以取消通过 subscribe(Collection）方式实现的订阅
 * 也可以取消通过 subscribe(Pattem）方式实现的订阅
 * 还可以取消通过 assign(Collection）方式实现的订阅
 * <p>
 * 集合订 阅 的方式 subscribe(Collection）
 * 正则表达式订 阅 的方式 subscribe(Pattem）
 * 指定分区 的订阅 方式 assign(Collection）
 * 分表代表了 三种不同的订阅状态 ：
 * AUTO_TOPICS 、
 * AUTO PATTERN
 * USER ASSIGNED （如果没有订阅 ， 那么 订 阅状态为 NONE ）。
 * 这三 种状态是互斥的， 在一个消费者中只能使用其中的一种
 * <p>
 * 通过 subscribe（）方法订阅主题具有消费者自动再均衡的功能 ，在 多个消费者 的情况下可以 根据分区分配策略来自动分配各个消费者与分区的关系。
 * 通过 assign（）方法订阅分区时， 是不具备消费者自动均衡的功能的，
 * 两种类型 的 subscribe（）都有 ConsumerRebalanceListener 类型参数 的方法，而 assign（）方法却没有。
 */
public class ConsumerFastStart {

    public static final String topic = "test";
    public static final String brokerList = "localhost:9092";
    public static final String groupId = "myGroup";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        //设置消费者组ID，即组名称，值可自定义。组名称相同的消费者进程属于同一个消费者组。
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        //客户端ID
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.client.id.a");
        //定义消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //3. 设置消费者读取的主题名称可以设置多个
        /*kafkaConsumer.subscribe(Arrays.asList("topictest"));*/
        kafkaConsumer.subscribe(Collections.singletonList(topic));
        while (true) {
            /**
             * 拉取消息
             * 可以直接将 timeout 设置为 0 ， 这样 poll（）方法会立刻返回
             * 如果应用线程唯一 的工作就是从 Kafka 中拉取并消费消息，则可以将 这个参数设置为最大值 Long.MAX_VALUE
             *
             *  consumerRecords.iterator();
             *  consumerRecords.partitions();获取消息集中指定分区的消息
             *  consumerRecords.isEmpty();
             *  consumerRecords.count();
             */
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.println(record.value());
                System.out.println(record.topic());
                System.out.println(record.offset());
                System.out.println(record.key());
                System.out.println(record.partition());
            }
        }
    }


}
